package yuekao9.dwd;

import yuekao9.util.KafkaUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ReadOrderOrgBound {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("tms_ods"));
        streamSource.print();
        //3.从Kafka tms_ods主题筛选中转业务流程相关数据，根据业务过程拆分，写入Kafka不同事实主题。（5分）
        OutputTag<String> dwdboundinboundbean = new OutputTag<String>("dwdboundinboundbean") {
        };
        OutputTag<String> dwdboundsortbean = new OutputTag<String>("dwdboundsortbean") {
        };
        SingleOutputStreamOperator<String> process = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                String table = JSON.parseObject(s).getString("tb");
                String op = JSON.parseObject(s).getString("op");
                String status = JSON.parseObject(s).getJSONObject("data").getString("status");
                String sort_time = JSON.parseObject(s).getJSONObject("data").getString("sort_time");
                if ("order_org_bound".equals(table) ) {
                    if ("64001".equals(status)&& "c".equals(op)) {
                        context.output(dwdboundinboundbean, s);
                    }else if (sort_time == null && "64003".equals(status)) {
                        context.output(dwdboundsortbean, s);
                    }
                }
            }
        });
//        process.getSideOutput(dwdboundinboundbean).print();
//        process.getSideOutput(dwdboundsortbean).print();
//        process.getSideOutput(dwdboundinboundbean).addSink("dwdboundinboundbean");
//                process.getSideOutput(dwdboundsortbean).addSink(dwdboundsortbean);

        env.execute();
    }
}
